package com.minhang.adsb.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.minhang.adsb.constant.VersionEnum;
import com.minhang.adsb.entity.ServerConfig;
import com.minhang.adsb.mapper.AdsbDynamicPointMapper;
import com.minhang.adsb.mapper.ServerConfigMapper;
import com.minhang.adsb.model.AirTrackData;
import com.minhang.adsb.service.AdsbDynamicPointService;
import com.minhang.adsb.service.PushStrategy;
import lombok.extern.slf4j.Slf4j;
import okhttp3.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service
@Slf4j
public class AdsbDynamicPointServiceImpl extends ServiceImpl<AdsbDynamicPointMapper, AirTrackData>
        implements AdsbDynamicPointService {

    // 超时计数器（线程安全）


    @Autowired
    private ServerConfigMapper serverConfigMapper;

    private final Map<VersionEnum, PushStrategy> strategyMap;

    @Autowired
    public AdsbDynamicPointServiceImpl(
            V1HttpPushStrategy v1Strategy,
            V2ProcessedHttpStrategy v2Strategy,
            V3KafkaPushStrategy v3Strategy) {
        this.strategyMap = new HashMap<>();
        strategyMap.put(VersionEnum.V1, v1Strategy);
        strategyMap.put(VersionEnum.V2, v2Strategy);
        strategyMap.put(VersionEnum.V3, v3Strategy);
    }

    // 优化后的线程池配置
    private static final ExecutorService executor = Executors.newFixedThreadPool(50);


    @Override
    public void toSave(String jsonPayload) {
        getActiveServers().forEach(server ->
                executor.execute(() -> {
                    VersionEnum version = VersionEnum.fromValue(server.getPushVersion());
                    PushStrategy strategy = strategyMap.get(version);
                    if (strategy != null) {
                        try {
                            strategy.push(server, jsonPayload);
                        } catch (Exception e) {
                            log.info("推送失败: {}", e.getMessage());
                        }
                    } else {
                        log.warn("未知推送版本: {}", server.getPushVersion());
                    }
                })
        );
    }

    // 获取活跃服务器
    private List<ServerConfig> getActiveServers() {
        List<ServerConfig> serverConfigs = serverConfigMapper.selectList(
                new LambdaQueryWrapper<ServerConfig>().eq(ServerConfig::getIsActive, 1));
        //根据endpoint去重
        serverConfigs = new ArrayList<>(serverConfigs.stream()
                .collect(Collectors.toMap(ServerConfig::getEndpoint, e -> e, (e1, e2) -> e1))
                .values());
        return serverConfigs;
    }
}